/*
 * To change this template, choose Tools | Templates
 * and open the template in the editor.
 */

package homomorphism.set;

import homomorphism.HomomorphismBase;
import org.apache.commons.logging.Log;
import org.apache.commons.logging.LogFactory;
import org.apache.hadoop.conf.Configuration;
import org.apache.hadoop.io.Writable;
import org.apache.hadoop.mapred.ClusterStatus;
import org.apache.hadoop.mapred.JobClient;
import org.apache.hadoop.mapreduce.Job;

/**
 *
 * @author liuyu
 */
public class SetHomomorphismwrapper<InKey extends Writable,InVal extends Writable,OutKey extends Writable,OutVal extends Writable> extends HomomorphismBase implements SetHomomorphismInterface<InVal,OutVal> {

    private static final String MyUsage = "usage:  [input file path]  [output file path] <-M [:num]> <-R [:num]>-h  :[usage]";
    private static final Log LOG = LogFactory.getLog(SetHomomorphismwrapper.class);

    public static class MyMapper<InKey,InVal,OutKey,OutVal> extends SetMapper<InKey,InVal,OutKey,OutVal> {
    }

    public static class MyReducer<InKey,InVal,OutKey,OutVal> extends SetReducer<InKey,InVal,OutKey,OutVal> {
    }

    @Override
    public int run(String[] args) throws Exception {
        Configuration conf = getConf();
        JobClient client = new JobClient(conf);
        ClusterStatus cluster = client.getClusterStatus();

        Job job = new Job(conf);

        job.setJobName("First_phase");
        job.setJarByClass(this.getClass());

        job = initJobConf(job, MyMapper.class, MyReducer.class, args);
        if (job== null) {
            LOG.error(" job init  failed ");
            return -1;
        }

        int ret = job.waitForCompletion(true) ? 0 : 1;
        return ret;
    }

    public OutVal filter(InVal val) {
        throw new UnsupportedOperationException("Not implemented.");
    }

    public OutVal plus(OutVal a, OutVal b) {
        throw new UnsupportedOperationException("Not implemented.");
    }

    public Job initJobConf(Job job,
            Class UserMapperClass,
            Class UserReducerClass,
            String[] args) throws Exception {



        return job;

    }

}
